package cn.com.idmy.mq.listener;

import cn.com.idmy.mq.coder.MqMsgDecoder;
import cn.com.idmy.mq.model.MqEvent;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;

import java.lang.reflect.Method;
import java.util.Map;

@Slf4j
@RequiredArgsConstructor
public abstract class AbstractMsgListener {
    protected final Map<String, Map<String, Class<?>>> messageTypeMap;
    protected final Map<String, Map<String, Method>> methodMap;
    protected final Map<String, Map<String, Object>> beanMap;
    protected final Map<String, Map<String, MqMsgDecoder>> decoderMap;

    protected boolean invoke(MessageExt msg) {
        try {
            Object body = null;
            String topic = msg.getTopic();
            String tag = StrUtil.isEmpty(msg.getTags()) ? "*" : msg.getTags();
            if (CollUtil.isNotEmpty(messageTypeMap) && messageTypeMap.containsKey(topic) && CollUtil.isNotEmpty(messageTypeMap.get(topic))) {
                body = getMessageDecoder(topic, tag).decode(msg, getMessageType(topic, tag));
            }

            if (body == null) {
                throw new Exception("decode body is null!");
            }

            if (CollUtil.isNotEmpty(methodMap)
                    && methodMap.containsKey(topic)
                    && CollUtil.isNotEmpty(methodMap.get(topic))
                    && CollUtil.isNotEmpty(beanMap)
                    && beanMap.containsKey(topic)) {
                Method method = getMethod(topic, tag);
                Object bean = getBean(topic, tag);
                if (method.getParameters().length == 2) {
                    MqEvent event = MqEvent.builder().topic(topic).tag(tag).key(msg.getKeys()).msgId(msg.getMsgId()).build();
                    method.invoke(bean, body, event);
                } else {
                    method.invoke(bean, body);
                }
                return true;
            } else {
                throw new Exception("can't find method ! reconsume message! msgId: " + msg.getMsgId());
            }
        } catch (Throwable e) {
            Throwable t = ExceptionUtil.getRootCause(e);
            log.error("e:" + t.getMessage(), t);
            log.warn("consume fail , ask for re-consume , msg: {}", msg);
            return false;
        }
    }

    protected Class<?> getMessageType(String topic, String tag) {
        return messageTypeMap.get(topic).get(messageTypeMap.get(topic).containsKey(tag) ? tag : "*");
    }

    protected Method getMethod(String topic, String tag) {
        return methodMap.get(topic).get(methodMap.get(topic).containsKey(tag) ? tag : "*");
    }

    protected Object getBean(String topic, String tag) {
        return beanMap.get(topic).get(beanMap.get(topic).containsKey(tag) ? tag : "*");
    }

    protected MqMsgDecoder getMessageDecoder(String topic, String tag) {
        return decoderMap.get(topic).get(decoderMap.get(topic).containsKey(tag) ? tag : "*");
    }
}
